package topology;

import groovy.json.internal.ValueList;
import kafka.newstyle.Producer;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

public class AreaToJingWeiBolt extends BaseFunction {
    List<String> shengfenList;
    List<String> chengshiList;
    List<Double> jingduList;
    List<Double> weiduiList;

    @Override
    public void prepare(Map<String, Object> conf, TridentOperationContext context) {

        // 读取地址信息，初始化minList,maxList,addressList变量
        String filePath = Producer.class.getClassLoader().getResource("data/lng-lat-mapping.txt").getFile();
        File area_jingwei = new File(filePath);
        InputStreamReader isr = null;
        try {
            isr = new InputStreamReader(new FileInputStream(area_jingwei), StandardCharsets.UTF_8);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        BufferedReader br = new BufferedReader(isr);
        String lineTxt;
        try {
            while ((lineTxt = br.readLine()) != null) {
                String[] jingwei=lineTxt.split(",");
                for(int i=0;i<jingwei.length;i++) {
                    jingduList.add(Double.valueOf(jingwei[3]));
                    weiduiList.add(Double.valueOf(jingwei[2]));
                    shengfenList.add(jingwei[0]);
                    chengshiList.add(jingwei[1]);

                }
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        String shengfen=tuple.getString(1);
        String chengshi=tuple.getString(2);
        for (int i=0;i<chengshiList.size();i++)
        if (shengfen.equals(shengfenList.get(i)) && chengshi.equals(chengshiList.get(i))){

            collector.emit(new Values(weiduiList.get(i),jingduList.get(i)));

        }
    }
}
